/*
 * Copyright (c) 2006-2018, RT-Thread Development Team
 *
 * SPDX-License-Identifier: Apache-2.0
 *
 * Change Logs:
 * Date           Author       Notes
 * 2019-07-11     flybreak     the first version
 */
#include "tcpserver.h"
#include <sys/ioctl.h>
#include <aos/aos.h>
#include <assert.h>
#include <ulog/ulog.h>

#define DBG_TAG   "tcpserv"

static tcpclient_t tcpserver_add_cli(struct tcpserver *serv, int fd_cli)
{
    tcpclient_t client = NULL;
    unsigned long ul = 1;
    int ret;
    aos_assert(serv);

    LOGD(DBG_TAG, "client %d id connected", fd_cli);

    client = (tcpclient_t)calloc(1, sizeof(struct tcpclient));
    if (client == NULL)
    {
        LOGE(DBG_TAG, "client calloc failed!");
        goto __exit;
    }

    client->server = serv;
    client->sock = fd_cli;
    sem_init(&client->semaphore, 0, 0);
    if (&client->semaphore == NULL)
    {
        LOGE(DBG_TAG, "client event create failed!");
        goto __exit;
    }

    /* set socket to be non-blocking */
    ret = ioctlsocket(fd_cli, FIONBIO, (unsigned long *)&ul);
    if (ret < 0)
    {
        LOGE(DBG_TAG, "set socket non-blocking failed");
    }

    /* add new client to cli_list */
    serv->cli_list[fd_cli] = client;

    /* update fd_max */
    FD_SET(fd_cli, &serv->read_set);
    if (fd_cli > serv->fd_max)
    {
        serv->fd_max = fd_cli;
    }

    /* notify new client */
    mq_send(serv->messageque, (const char *)client, sizeof(client), 0);
    if (serv->tcpserver_event_notify)
    {
        serv->tcpserver_event_notify(client, TCPSERVER_EVENT_CONNECT);
    }
    return client;

__exit:
    if (&client->semaphore)
        sem_destroy(&client->semaphore);
    if (client)
        free(client);

    return NULL;
}

static void tcpserver_del_cli(tcpclient_t client)
{
    int max = -1;
    int i = 0;
    struct tcpserver *serv = client->server;

    aos_assert(client);

    /* remove client from read_set */
    FD_CLR(client->sock, &serv->read_set);

    /* update fd_max */
    for (i = 0; i <= serv->fd_max; ++i)
    {
        if (!FD_ISSET(i, &serv->read_set))
        {
            continue;
        }
        if (max < i)
        {
            max = i;
        }
    }
    serv->fd_max = max;

    /* close client socket */
    closesocket(client->sock);

    /* remove client from client_list */
    serv->cli_list[client->sock] = NULL;

    /* notify disconnect */
    if (serv->tcpserver_event_notify)
    {
        serv->tcpserver_event_notify(client, TCPSERVER_EVENT_DISCONNECT);
    }

    /* free memory */
    sem_destroy(&client->semaphore);
    free(client);
    LOGD(DBG_TAG, "socket:%d,closed", client->sock);
}

static void *tcpserver_thread_entry(void *parameter)
{
    struct tcpserver *server = parameter;
    fd_set  read_set, write_set;
    int ret_sel, ret;
    unsigned long ul = 1;
    struct timeval time = {0};

    server->state = TCPSERVER_STATE_RUN;
    server->fd_max = server->sock;
    time.tv_sec =  1;
    time.tv_usec = 0;

    /* set socket to be non-blocking */
    ret = ioctlsocket(server->sock, FIONBIO, (unsigned long *)&ul);
    if (ret < 0)
    {
        LOGE(DBG_TAG, "set socket ctl failed");
    }

    FD_ZERO(&server->read_set);
    FD_ZERO(&server->write_set);
    FD_SET(server->sock, &server->read_set);

    while (1)
    {
        read_set = server->read_set;
        write_set = server->write_set;

        ret_sel = select(server->fd_max + 1, &read_set, &write_set, NULL, (void *)&time);

        /* detection stop mark */
        if (server->state == TCPSERVER_STATE_STOP)
        {
            LOGD(DBG_TAG, "server thread exit.");
            return NULL;
        }
        /* select error or timeout */
        if (ret_sel <= 0)
        {
            continue;
        }

        if (FD_ISSET(server->sock, &read_set))
        {
            /* accept new client */
            struct sockaddr_in peer;
            int fd_cli_new;
            socklen_t len;

            len = sizeof(peer);

            fd_cli_new = accept(server->sock, (struct sockaddr *)&peer, &len);
            if (fd_cli_new < 0)
            {
                LOGE(DBG_TAG, "accept error");
                continue;
            }
            /* add client */
            tcpserver_add_cli(server, fd_cli_new);
        }
        else    /*  */
        {
            /* received data */
            int i = 0;

            for (i = 0; i <= server->fd_max; i++)
            {
                if (!FD_ISSET(i, &read_set))
                {
                    continue;
                }
                if (server->cli_list[i])
                {
                    sem_post(&server->cli_list[i]->semaphore);
                    if (server->tcpserver_event_notify)
                    {
                        server->tcpserver_event_notify(server->cli_list[i], TCPSERVER_EVENT_RECV);
                    }
                }
            }
        }
    }
}

size_t tcpserver_send(tcpclient_t client, void *buf, size_t size, int32_t timeout)
{
    return send(client->sock, buf, size, 0);
}

size_t tcpserver_recv(tcpclient_t client, void *buf, size_t size, int32_t timeout)
{
    uint32_t e;
    int32_t recv_size;
    aos_assert(client);
    aos_assert(buf != NULL);

    sem_wait(&client->semaphore);
    {
        recv_size = recv(client->sock, buf, size, 0);

        if (recv_size > 0)
        {
            return recv_size;
        }
        else if (recv_size == 0)
        {
            LOGD(DBG_TAG, "recv 0");
            tcpserver_del_cli(client);
        }
        else if (recv_size < 0)
        {
            if (!(recv_size == EINTR || recv_size == EWOULDBLOCK || recv_size == EAGAIN))
            {
                LOGE(DBG_TAG, "recv error");
                tcpserver_del_cli(client);
            }
        }
    }
    return 0;
}

tcpclient_t tcpserver_accept(struct tcpserver *server, int32_t timeout)
{
    tcpclient_t cli;

    aos_assert(server != NULL);

    if (mq_receive(server->messageque, (char *)&cli, sizeof(cli), NULL) > 0)
    {
        return cli;
    }
    else
    {
        return NULL;
    }
}

int tcpserver_close(tcpclient_t client)
{
    tcpserver_del_cli(client);
    return 0;
}

void tcpserver_set_notify_callback(struct tcpserver *server, void (*tcpserver_event_notify)(tcpclient_t client, uint8_t event))
{
    server->tcpserver_event_notify = tcpserver_event_notify;
}

static int tcpserver_start(struct tcpserver *server)
{
    struct sockaddr_in addr;
    int ret_bind, ret_listen;
    int opt = 1;

    server->sock = socket(AF_INET, SOCK_STREAM, 0);
    if (server->sock < 0)
    {
        LOGE(DBG_TAG, "socket create failed");
        return -1;
    }

    if (server->ip)
        addr.sin_addr.s_addr = inet_addr(server->ip);
    else
        addr.sin_addr.s_addr = INADDR_ANY;
    addr.sin_family = AF_INET;
    addr.sin_port = htons(server->port);

    /* set server socket port multiplexing */
    setsockopt(server->sock, SOL_SOCKET, SO_REUSEADDR, (const void *)&opt, sizeof(opt));

    ret_bind = bind(server->sock, (struct sockaddr *)&addr, sizeof(addr));
    if (ret_bind < 0)
    {
        LOGE(DBG_TAG, "bind failed");
        goto __exit;
    }

    ret_listen = listen(server->sock, TCPSERVER_CLI_NUM);
    if (ret_listen < 0)
    {
        LOGE(DBG_TAG, "listen failed");
        goto __exit;
    }

    server->thread = malloc(sizeof(server->thread));
    if (NULL == server->thread)
    {
        LOGE(DBG_TAG, "no memory malloc thread");
        goto __exit;
    }

    int res = pthread_create(&server->thread, NULL, tcpserver_thread_entry, server);
    if (0 != res)
    {
        LOGE(DBG_TAG, "thread create failed");
        goto __exit;
    }
    return 0;
__exit:
    if (server->sock)
        closesocket(server->sock);
    return -1;
}

struct tcpserver *tcpserver_create(const char *ip, uint16_t port)
{
    struct tcpserver *server = NULL;

    server = (struct tcpserver *)calloc(1, sizeof(struct tcpserver));
    if (server == NULL)
    {
        LOGE(DBG_TAG, "no memory for tcp server");
        return NULL;
    }

    struct mq_attr attr = {0};
    attr.mq_maxmsg = TCPSERVER_CLI_NUM;
    attr.mq_msgsize = sizeof(tcpclient_t);

    mq_unlink(TCPSERVER_NAME);
    server->messageque = mq_open(TCPSERVER_NAME, O_CREAT | O_RDWR, 0600, &attr);
    if (server->messageque == (mqd_t) -1)
    {
        LOGE(DBG_TAG, "no memory for messageque");
        goto __exit;
    }
    server->cli_list = (tcpclient_t *)calloc(1, sizeof(tcpclient_t) * TCPSERVER_SOCKET_MAX);
    if (server->cli_list == NULL)
    {
        LOGE(DBG_TAG, "no memory for cli_list");
        goto __exit;
    }

    server->ip = ip;
    server->port = port;

    if (tcpserver_start(server) != 0)
    {
        LOGE(DBG_TAG, "tcp server start failed");
        goto __exit;
    }

    return server;

__exit:
    LOGE(DBG_TAG, "error!");
    if (server->messageque)
        mq_close(server->messageque);
    if (server->cli_list)
        free(server->cli_list);
    if (server)
        free(server);
    return NULL;
}

int tcpserver_destroy(struct tcpserver *server)
{
    int i;

    /* wait for the select thread to exit */
    server->state = TCPSERVER_STATE_STOP;

    /* close all clients */
    for (i = 0; i <= server->fd_max; i++)
    {
        if (!FD_ISSET(i, &server->read_set))
        {
            continue;
        }
        if (i != server->sock)
        {
            tcpserver_close(server->cli_list[i]);
        }
    }

    /* close server socket */
    closesocket(server->sock);

    /* free memory */
    if (server->messageque)
        mq_close(server->messageque);
    if (server->cli_list)
        free(server->cli_list);
    if (server)
        free(server);

    return 0;
}
